package com.haiyou.data.common.accessor.caffeine;

import java.io.Serializable;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;

import org.checkerframework.checker.index.qual.NonNegative;
import org.checkerframework.checker.nullness.qual.NonNull;
import org.checkerframework.checker.nullness.qual.Nullable;

import com.alibaba.fastjson2.util.TypeUtils;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.Expiry;
import com.github.benmanes.caffeine.cache.RemovalCause;
import com.google.common.collect.Iterables;
import com.haiyou.common.entity.Entity;
import com.haiyou.common.entity.KeyPrefixChecker;
import com.haiyou.common.entity.Prefix;
import com.haiyou.common.id.IdCreator;
import com.haiyou.common.thread.AtomicExecutor;
import com.haiyou.common.util.reflect.AnnotationFinder;
import com.haiyou.common.util.string.StringUtils;
import com.haiyou.data.common.accessor.AbstractAccessor;
import com.haiyou.data.common.accessor.rsync.Synchronizer;
import com.haiyou.data.common.cache.redis.RKeys;
import com.haiyou.data.common.persister.mysql.persister.Persister;

import lombok.Getter;

/**
 * 本地缓存+mysql存储器，仅仅适用于单进程业务，对于更新的操作请使用自带的compute函数，切勿使用临时变量进行各种set操作，
 * 避免不能和mysql保持一致
 * 
 * @author xingyuan
 *
 * @param <K>
 * @param <V>
 */
public abstract class CaffeineAccessor<K extends Serializable & Comparable<K>, V extends Entity<K>>
		extends AbstractAccessor<K, V> {

	/**
	 * 缓存操作类型
	 * 
	 * @author Administrator
	 *
	 */
	public enum Type {
		AFTER_CREATE, AFTER_UPDATE, AFTER_READ,;
	}

	protected Cache<K, V> cache;

	protected AtomicLong maxId;

	@Getter
	protected Persister<K, V> persister;

	private Synchronizer<K, V> synchronizer;

	private Expiry<K, V> expiry = new Expiry<K, V>() {
		@Override
		public long expireAfterCreate(@NonNull K key, @NonNull V value, long currentTime) {
			return timeToLive(Type.AFTER_CREATE, key, value);
		};

		@Override
		public long expireAfterUpdate(@NonNull K key, @NonNull V value, long currentTime,
				@NonNegative long currentDuration) {
			return timeToLive(Type.AFTER_UPDATE, key, value);
		};

		@Override
		public long expireAfterRead(@NonNull K key, @NonNull V value, long currentTime,
				@NonNegative long currentDuration) {
			return timeToLive(Type.AFTER_READ, key, value);
		};
	};

	protected long maxPk() {
		K maxPk = persister.maxPk();
		if (maxPk == null) {
			return IdCreator.create();
		}
		String maxStr = maxPk.toString();
		if (StringUtils.isAllNumber(maxStr)) {
			return Long.parseLong(maxStr);
		} else {
			String[] ss = maxStr.split("_");
			if (ss == null || ss.length < 1) {
				return IdCreator.create();
			}
			String v = ss[ss.length - 1];
			if (StringUtils.isEmpty(v)) {
				return IdCreator.create();
			}
			if (StringUtils.isAllNumber(v)) {
				return Long.parseLong(v);
			} else {
				return IdCreator.create();
			}
		}
	}

	/**
	 * the length of time before the entry expires, in nanoseconds
	 * 
	 * <pre>
	 * 
	 * 过期前的时间长度，单位-纳秒，重写该方法可以针对单个对象修改过期时间
	 * 
	 * @param key
	 * @param value
	 * @return
	 */
	protected long timeToLive(Type type, K key, V value) {
		return synchronizer.getTimeToLiveUnit().toNanos(synchronizer.getTimeToLive());
	}

	@Override
	public void init() {

		super.init();

		persister = Persister.getInstance(valueClass);

		maxId = new AtomicLong(maxPk());

		synchronizer = new Synchronizer<>(this, persister);
		synchronizer.init();

		@NonNull
		Caffeine<Object, Object> caffeine = Caffeine.newBuilder();
		if (synchronizer.getInitialCapacity() > 0)
			caffeine.initialCapacity(synchronizer.getInitialCapacity());
		if (synchronizer.getMaximumSize() > 0)
			caffeine.maximumSize(synchronizer.getMaximumSize());
		if (synchronizer.getTimeToLive() > 0)
			caffeine.expireAfter(expiry);
		caffeine.removalListener((@Nullable K key, @Nullable V value, @NonNull RemovalCause cause) -> {
			if (cause != RemovalCause.EXPLICIT && value != null && key != null)
				synchronizer.flush(key, value);
		});
		cache = caffeine.build();
	}

	@Override
	public void destory() {
		super.destory();
		synchronizer.destory();
	}

	@Override
	public K generateKey() {
		K id = super.generateKey();
		if (id != null)
			return id;
		long incred = maxId.incrementAndGet();
		if (keyClass.getName().compareTo(String.class.getName()) == 0) {
			Prefix prefix = AnnotationFinder.findAnnotation(getClass(), Prefix.class);
			KeyPrefixChecker.checkFlag(prefix.value());
			String keyPrefix = prefix.value();
			return id = (K) RKeys.createRId(keyPrefix, incred);
		}
		return id = TypeUtils.cast(incred, keyClass);
	}

	@Override
	public V generate(Consumer<V> consumer) {
		K k = null;
		V v = null;
		try {
			k = generateKey();
			v = valueClass.newInstance();
			v.setId(k);
			if (consumer != null)
				consumer.accept(v);
			put(k, v);
		} catch (Exception e) {
			throw new RuntimeException(e.getMessage(), e);
		}
		return v;
	}

	@Override
	public Map<K, V> getAll(Iterable<? extends K> keys) {
		return cache.getAll(keys,
				ids -> persister.queryBatch(ids).stream().collect(Collectors.toMap(V::getId, n -> n, (n, m) -> m)));
	}

	@Override
	public boolean containsKey(Object key) {
		return cache.getIfPresent((K)key) != null || persister.contains((K) key);
	}

	@Override
	public boolean containsValue(Object value) {
		return get(((V) value).getId()).equals(value);
	}

	@Override
	public V get(Object key) {
		return cache.get((K) key, id -> persister.query(id));
	}

	@Override
	public V put(K key, V value) {
		try {
			cache.put(key, value);
			return cache.getIfPresent(key);
		} finally {
			synchronizer.flush(key, value);
		}
	}

	@Override
	public V remove(Object key) {
		V old = cache.get((K) key, id -> persister.query(id));
		cache.invalidate((K)key);
		synchronizer.giveUp((K) key);
		persister.delete((K) key);
		if (keyRecycler != null)
			keyRecycler.retrieve((K) key);
		return old;
	}

	@Override
	public void removeAll(Iterable<? extends K> keys) {
		cache.invalidateAll(keys);
		synchronizer.giveUp(keys);
		persister.delete(keys);
		if (keyRecycler != null)
			keyRecycler.retrieve(Iterables.toArray(keys, keyClass));
	}

	@Override
	public void putAll(Map<? extends K, ? extends V> m) {
		cache.putAll(m);
		synchronizer.flush(m);
	}

	@Override
	public V putIfAbsent(K key, V value) {
		return AtomicExecutor.call(key, () -> super.putIfAbsent(key, value));
	}

	@Override
	public boolean remove(Object key, Object value) {
		AtomicBoolean bool = new AtomicBoolean();
		AtomicExecutor.run((K) key, () -> bool.compareAndSet(false, super.remove(key, value)));
		return bool.get();
	}

	@Override
	public boolean replace(K key, V oldValue, V newValue) {
		AtomicBoolean bool = new AtomicBoolean();
		AtomicExecutor.run((String) key, () -> bool.compareAndSet(false, super.replace(key, oldValue, newValue)));
		return bool.get();
	}

	@Override
	public V replace(K key, V value) {
		return AtomicExecutor.call(key, () -> super.replace(key, value));
	}

	@Override
	public V computeIfAbsent(K key, Function<? super K, ? extends V> mappingFunction) {
		return AtomicExecutor.call(key, () -> super.computeIfAbsent(key, mappingFunction));
	}

	@Override
	public V computeIfPresent(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
		return AtomicExecutor.call(key, () -> super.computeIfPresent(key, remappingFunction));
	}

	@Override
	public V compute(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
		return AtomicExecutor.call(key, () -> super.compute(key, remappingFunction));
	}

	@Override
	public V merge(K key, V value, BiFunction<? super V, ? super V, ? extends V> remappingFunction) {
		return AtomicExecutor.call(key, () -> super.merge(key, value, remappingFunction));
	}
}
